package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(Command interface{}) (index, term, isleader)
//   start agreement on a new Log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the Log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"6.824/labgob"
	"bytes"
	"log"
	"math/rand"
	"sort"

	//	"bytes"
	"sync"
	"sync/atomic"
	"time"

	//	"6.824/labgob"
	"6.824/labrpc"
)

const Candidates = 0
const Followers = 1
const Leaders = 2

//
// as each Raft peer becomes aware that successive Log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed Log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's State
	cond      sync.Cond           // Lock to protect shared access to this peer's State
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted State
	applyCh   chan ApplyMsg       // Object to hold this peer's persisted State
	Me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	CurrentTerm  int
	VotedFor     int
	Log          []LogEntry
	snapshot     []byte
	includeIndex int
	includeTerm  int

	State     int
	Heartbeat bool

	//voteArgChan chan RequestVoteArgs
	//voteChan    chan RequestVoteArgs

	snapChan  chan InstallSnapshotArgs
	heartChan chan AppendEntriesArgs

	CommitIndex int
	LastApplied int

	NextIndex  []int
	MatchIndex []int
	// Look at the paper's Figure 2 for a description of what
	// State a Raft server must maintain.

}

type sql struct {
	Term         int
	Vote         int
	Log          []LogEntry
	Snapshot     []byte
	IncludeIndex int
	IncludeTerm  int
}
type snapshot struct {
	Snapshot     []byte
	IncludeIndex int
	IncludeTerm  int
}

type LogEntry struct {
	Index   int
	Term    int
	Command interface{}
}

func (rf *Raft) setTerm(t int) {
	//rf.mu.Lock()
	//defer rf.mu.Unlock()
	if rf.CurrentTerm != t {
		rf.VotedFor = -1
	}
	rf.CurrentTerm = t
}

//type PState

// return CurrentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	var term int
	var isleader bool
	// Your code here (2A).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	term = rf.CurrentTerm
	isleader = rf.State == Leaders
	return term, isleader
}

//
// save Raft's persistent State to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)

	now := time.Now()
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)

	per := sql{rf.CurrentTerm, rf.VotedFor,
		rf.Log, []byte{}, rf.includeIndex, rf.includeTerm}
	//DPrintf("me[%v] save [%v]", rf.Me, per)
	defer func() {
		DPrintf("persist cost[%v] sql[%v]", time.Since(now), rf.Me)
	}()

	err := e.Encode(per)
	if err != nil {
		DPrintf("persist state err[%v]", err)
		return
	}
	//e.Encode(rf.yyy)
	//data := w.Bytes()
	rf.persister.SaveStateAndSnapshot(w.Bytes(), rf.snapshot)
}

//
// restore previously persisted State.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any State?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var xxx sql
	err := d.Decode(&xxx)
	if err != nil {
		log.Println(err)
		return
	}

	rf.VotedFor = xxx.Vote
	rf.CurrentTerm = xxx.Term
	rf.Log = xxx.Log
	rf.snapshot = xxx.Snapshot
	rf.includeIndex = xxx.IncludeIndex
	rf.includeTerm = xxx.IncludeTerm
	rf.LastApplied = xxx.IncludeIndex

}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).
	//DPrintf("1 cond [%v]", rf.Me)
	rf.mu.Lock()
	defer rf.mu.Unlock()
	DPrintf("2 cond [%v]", rf.Me)

	//last := getLog(rf.LastApplied, rf.log)
	if rf.includeTerm > lastIncludedTerm || rf.includeIndex > lastIncludedIndex {
		DPrintf("3 cond [%v]", rf.Me)
		return false
	}
	rf.Log = copyLogHigher(lastIncludedIndex+1, rf.Log)
	rf.LastApplied = lastIncludedIndex
	rf.snapshot = snapshot
	rf.includeIndex = lastIncludedIndex
	rf.includeTerm = lastIncludedTerm
	rf.Heartbeat = true
	rf.persist()

	//DPrintf("4 cond [%v]", rf)
	return true
}

const MAXLOGSIZE = 1000

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	//rf.mu.Lock()
	//defer rf.mu.Unlock()
	//rf.log = copyLogHigher(index, rf.log)
	//rf.persist()
	//if rf.State != Leaders {
	//	return
	//}

	rf.mu.Lock()
	defer rf.mu.Unlock()
	rf.Log = copyLogHigher(index, rf.Log)
	logEntry := rf.Log[0]
	DPrintf("Snapshot me[%v] index[%v]  rf.Log[%v]", rf.Me, index, rf.Log)
	rf.Log = rf.Log[1:]
	rf.snapshot = snapshot
	rf.includeIndex = logEntry.Index
	rf.includeTerm = logEntry.Term
	rf.LastApplied = rf.includeIndex
	rf.persist()
	//rf.log = make([]LogEntry, 0)
	//rf.log = append(rf.log, LogEntry{})
}

type InstallSnapshotArgs struct {
	Term             int
	LeaderId         int
	LastIncludeIndex int
	LastIncludeTerm  int
	Snapshot         []byte

	server int
}

type InstallSnapshotReply struct {
	Term int
}

func (rf *Raft) InstallSnapshot(arg *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	rf.mu.Lock()

	reply.Term = rf.CurrentTerm
	if arg.Term < reply.Term {
		rf.mu.Unlock()

		return
	}
	rf.doChangeFlower(arg.Term)

	rf.mu.Unlock()
	DPrintf("InstallSnapshot me[%v], arg[%v]", rf.Me, arg)
	rf.applyCh <- ApplyMsg{SnapshotValid: true, Snapshot: arg.Snapshot,
		SnapshotTerm: arg.LastIncludeTerm, SnapshotIndex: arg.LastIncludeIndex}

}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

func (rf *Raft) doChangeFlower(term int) {
	if term <= rf.CurrentTerm {
		return
	}
	rf.setTerm(term)
	rf.State = Followers
	rf.applyCh <- ApplyMsg{}
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	//marshal, _ := json.Marshal(rf)
	//DPrintf("rece vote,args[%v], rf[%v]",args, string(marshal))

	if args.Term < rf.CurrentTerm {
		return
	}

	if args.Term > rf.CurrentTerm {
		rf.doChangeFlower(args.Term)
	}
	reply.Term = rf.CurrentTerm
	last := rf.getLastLog()
	reply.VoteGranted =
		(rf.VotedFor == -1 || rf.VotedFor == args.CandidateId) &&
			//(!(rf.Log[len(rf.Log)-1].Term > args.LastLogTerm || len(rf.Log)-1 > args.LastLogIndex))
			!(args.LastLogTerm < last.Term || args.LastLogTerm == last.Term && last.Index > args.LastLogIndex)
	//&& args.LastLogIndex > len(rf.Log))
	if reply.VoteGranted {
		rf.VotedFor = args.CandidateId
		rf.Heartbeat = true
	}
	rf.persist()
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	//DPrintf("1 vote from[%v], to[%v], arg[%v], apply[%v]\n", args.CandidateId, server, args, reply)
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	DPrintf("2 vote from[%v], to[%v], arg[%v], apply[%v]\n", args.CandidateId, server, args, reply)
	return ok
}

type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []LogEntry
	//src          []LogEntry
	LeaderCommit int

	server int
}
type AppendEntriesReply struct {
	Term    int
	Success bool
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	//DPrintf("1 heat from[%v], to[%v], arg[%v], apply[%v],rf[]\n", args.LeaderId, server, args, reply)
	now := time.Now()
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)

	DPrintf("cost[%v] ok[%v] heat from[%v], to[%v], arg[%v], apply[%v],rf[]\n", time.Now().Sub(now), ok, args.LeaderId, server, args, reply)

	return ok
}

func getLog(index int, log []LogEntry) *LogEntry {
	if len(log) <= 0 || log[0].Index > index || log[len(log)-1].Index < index {
		return nil
	}

	for _, v := range log {
		if v.Index == index {
			return &v
		}
	}
	return nil
}
func copyLogLower(index int, log []LogEntry) []LogEntry {
	for i, v := range log {
		if v.Index == index {
			return log[:i]
		}
	}
	return []LogEntry{}
}
func copyLogHigher(index int, log []LogEntry) []LogEntry {
	entries := []LogEntry{}
	for _, v := range log {
		if v.Index >= index {
			entries = append(entries, v)
		}
	}
	return entries
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {

	now := time.Now()
	rf.mu.Lock()
	defer rf.mu.Unlock()

	rf.Heartbeat = true
	defer func() {
		DPrintf("AppendEntries me[%v] cost[%v] ", rf.Me, time.Now().Sub(now))
	}()
	if args.Term >= rf.CurrentTerm {
		rf.setTerm(args.Term)
		rf.State = Followers
	}
	reply.Term = rf.CurrentTerm
	if rf.State != Followers {
		return
	}
	if reply.Term > args.Term {
		return
	}
	prevLogIndex := args.PrevLogIndex

	prevLog := getLog(prevLogIndex, rf.Log)
	if prevLog == nil {
		prevLog = &LogEntry{rf.includeIndex, rf.includeTerm, ""}
	}
	if prevLog.Term != args.PrevLogTerm || prevLog.Index != args.PrevLogIndex {
		return
	}

	i := 0
	for j, v := range args.Entries {
		for i < len(rf.Log) && rf.Log[i].Index < v.Index {
			i++
		}
		if i >= len(rf.Log) {
			rf.Log = append(rf.Log, args.Entries[j:]...)
			break
		} else if rf.Log[i].Term != v.Term {
			rf.Log = append(rf.Log[:i], args.Entries[j:]...)
			break
		}
	}

	reply.Success = true
	logEntry := rf.getLastLog()

	if args.LeaderCommit >= rf.CommitIndex {
		rf.commitLog(0, min(args.LeaderCommit, logEntry.Index))
	}
	rf.persist()

	//DPrintf("append rf[%v]  reply[%v]", rf, reply)
	//DPrintf("append")
	//return true

}

func (rf *Raft) getLastLog() LogEntry {
	logEntry := LogEntry{rf.includeIndex, rf.includeTerm, ""}
	if len(rf.Log) > 0 {
		logEntry = rf.Log[len(rf.Log)-1]
	}
	return logEntry
}

func (rf *Raft) sendSingle(args AppendEntriesArgs) {

	x := args.server
	//marshal, _ := json.Marshal(rf)
	//DPrintf("send rf[%v]",string(marshal))
	reply := &AppendEntriesReply{}
	//now := time.Now()
	//args.Entries = copyLogHigher(args.PrevLogIndex+1, args.src)
	ok := rf.sendAppendEntries(x, &args, reply)

	rf.mu.Lock()

	if !ok || args.Term != rf.CurrentTerm {
		rf.mu.Unlock()
		return
	}

	if reply.Term > rf.CurrentTerm {
		rf.doChangeFlower(reply.Term)
		rf.mu.Unlock()
		return
	}
	DPrintf("if reply.Success[%v]", reply.Success)
	if reply.Success {
		last := LogEntry{Term: args.PrevLogTerm, Index: args.PrevLogIndex}
		if len(args.Entries) > 0 {
			last = args.Entries[len(args.Entries)-1]
		}

		rf.MatchIndex[x] = max(rf.MatchIndex[x], last.Index)
		rf.NextIndex[x] = max(rf.NextIndex[x], rf.MatchIndex[x]+1)
		//rf.MatchIndex[x] = last.Index
		//rf.NextIndex[x] = rf.MatchIndex[x] + 1
		l := rf.CommitIndex
		arr := make([]int, 0)
		for _, v := range rf.MatchIndex {
			arr = append(arr, v)
		}
		sort.Ints(arr)
		DPrintf("match arr[%v]", arr)
		nextC := arr[len(arr)/2]
		entry := getLog(nextC, rf.Log)
		if entry == nil {
			entry = &LogEntry{rf.includeTerm, rf.includeIndex, ""}
		}

		//if entry.Index == nextC && entry.Term == rf.CurrentTerm {
		if entry.Index == nextC {
			if nextC != rf.CommitIndex {
				defer rf.sendAppendEntriesAll()
			}
			rf.commitLog(l, nextC)

		}

	} else {
		rf.snapChan <- InstallSnapshotArgs{rf.CurrentTerm, rf.Me,
			rf.includeIndex, rf.includeTerm, rf.snapshot, args.server}
	}

	rf.persist()
	rf.mu.Unlock()

}

func (rf *Raft) sendAppendEntriesAll() bool {
	defer func() {
		rf.mu.Unlock()
	}()
	rf.mu.Lock()
	DPrintf("time sleep me[%v]", rf.Me)

	if rf.State != Leaders || rf.killed() {
		return false
	}

	//DPrintf("start[%v]", time.Now())
	for i := range rf.peers {
		if i == rf.Me {
			rf.Heartbeat = true
			rf.MatchIndex[i] = rf.getLastLog().Index
			continue
		}

		prevId := rf.NextIndex[i] - 1
		if prevId < rf.includeIndex {
			args := InstallSnapshotArgs{rf.CurrentTerm, rf.Me,
				rf.includeIndex, rf.includeTerm, rf.snapshot, i}
			rf.snapChan <- args
		} else {
			var prev *LogEntry
			if prevId == rf.includeIndex {
				prev = &LogEntry{prevId, rf.includeTerm, ""}
			} else {
				prev = getLog(prevId, rf.Log)
			}

			args := AppendEntriesArgs{}
			args.Term = rf.CurrentTerm
			args.LeaderId = rf.Me
			args.LeaderCommit = rf.CommitIndex
			args.PrevLogIndex = prev.Index
			args.PrevLogTerm = prev.Term
			args.Entries = copyLogHigher(args.PrevLogIndex+1, rf.Log)
			args.server = i
			//now := time.Now()
			rf.heartChan <- args
			//DPrintf("heartChan cost[%v] server[%v]", time.Now().Sub(now), i)

		}
	}
	//DPrintf("end[%v]", time.Now())

	return true
}

func (rf *Raft) sendSnap(args InstallSnapshotArgs) {

	reply := InstallSnapshotReply{}
	x := args.server
	now := time.Now()
	ok := rf.peers[x].Call("Raft.InstallSnapshot", &args, &reply)
	DPrintf("ok[%v] sendSnap cost[%v] from[%v], to[%v], arg[%v], apply[%v],rf[]\n", ok, time.Now().Sub(now), args.LeaderId, x, args, reply)

	if !ok {
		return
	}
	rf.mu.Lock()
	defer rf.mu.Unlock()
	rf.doChangeFlower(reply.Term)
	rf.MatchIndex[x] = args.LastIncludeIndex
	rf.NextIndex[x] = rf.MatchIndex[x] + 1
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next Command to be appended to Raft's Log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// Command will ever be committed to the Raft Log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the Command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func max(a int, b int) int {
	if b > a {
		a = b
	}
	return a
}

func min(a int, b int) int {
	if b < a {
		a = b
	}
	return a
}

func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	defer rf.sendAppendEntriesAll()
	defer rf.mu.Unlock()
	index := 0
	// Your code here (2B).
	if len(rf.Log) > 0 {
		index = rf.Log[len(rf.Log)-1].Index + 1
	} else {
		index = rf.includeIndex + 1
	}
	term := rf.CurrentTerm
	isLeader := rf.State == Leaders
	if !isLeader {
		return index, term, isLeader
	}

	DPrintf("recv command[%v], me[%v] index[%v] term[%v]\n", command, rf.Me, index, rf.CurrentTerm)
	//DPrintf("do command[%v], me[%v]\n", command, rf.Me)

	rf.Log = append(rf.Log, LogEntry{Term: rf.CurrentTerm, Command: command, Index: index})
	rf.persist()

	return index, term, isLeader
}

//DPrintf("cond.Wait()")

func (rf *Raft) commitLog(l int, r int) {
	rf.CommitIndex = max(r, rf.CommitIndex)
	rf.cond.Broadcast()
}

func (rf *Raft) commitWorker() {

	for !rf.killed() {
		//time.Sleep(50 * time.Millisecond)
		rf.cond.L.Lock()
		now := time.Now()
		for time.Since(now) < 50*time.Millisecond {
			i := rf.LastApplied - rf.includeIndex
			if i < 0 || i >= len(rf.Log) {
				break
			}
			v := rf.Log[i]
			if v.Index > rf.CommitIndex {
				break
			}
			msg := ApplyMsg{CommandValid: true, Command: v.Command, CommandIndex: v.Index}

			select {
			case rf.applyCh <- msg:
				DPrintf("cmid[%v] last apply[%v] commit me[%v] msg[%v] log[%v]", rf.CommitIndex, rf.LastApplied, rf.Me, msg, make([]string, 0))
				rf.LastApplied = max(rf.LastApplied, msg.CommandIndex)
			case <-time.After(10 * time.Millisecond):
				DPrintf("block v.index [%v] me[%v]", v.Index, rf.Me)
				rf.cond.Wait()
				//rf.mu.Unlock()
				//rf.mu.Lock()
				DPrintf("block return [%v] me[%v]", v.Index, rf.Me)
			}
		}
		DPrintf("me[%v] commit done cmit[%v]  rf.LastApplied[%v]", rf.Me, rf.CommitIndex, rf.LastApplied)
		rf.cond.Wait()
		rf.cond.L.Unlock()

	}

}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
	DPrintf("[%v] killed", rf.Me)
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

// The ticker go routine starts a new election if this peer hasn't received
// heartsbeats recently.
func (rf *Raft) ticker() {
	for rf.killed() == false {

		getHeartBeatTimeout()

		rf.doSelect()
	}
}

func getHeartBeatTimeout() {
	time.Sleep(time.Duration(1000+rand.Int31n(500)) * time.Millisecond)
}

//func (rf *Raft)

const heart = 300

func (rf *Raft) doSelect() {

	rf.mu.Lock()
	defer func() {
		rf.Heartbeat = false
		rf.mu.Unlock()
	}()
	//marshal, _ := json.Marshal(rf)
	//DPrintf("doSelect[%v]", string(marshal))
	if rf.Heartbeat {
		return
	}
	rf.State = Candidates
	rf.CurrentTerm++
	rf.VotedFor = rf.Me
	rf.persist()

	agree := 0

	lastLog := rf.getLastLog()
	args := &RequestVoteArgs{rf.CurrentTerm, rf.Me, lastLog.Index, lastLog.Term}
	//DPrintf("agree[%v]", agree)
	for i := range rf.peers {
		if i == rf.Me {
			continue
		}
		go func(x int) {
			reply := &RequestVoteReply{}

			//DPrintf("vote from [%v]  to[%v]", args.CandidateId, x)
			rf.sendRequestVote(x, args, reply)
			if !reply.VoteGranted && reply.Term <= args.Term {
				return
			}

			defer rf.mu.Unlock()
			rf.mu.Lock()
			if reply.Term > rf.CurrentTerm {
				rf.doChangeFlower(reply.Term)
				return
			}
			if rf.State != Candidates {
				return
			}
			agree++
			if agree+1 > len(rf.peers)/2 {
				DPrintf("%v is Leader at %v", rf.Me, rf.CurrentTerm)
				rf.State = Leaders
				rf.NextIndex = make([]int, len(rf.peers))
				rf.MatchIndex = make([]int, len(rf.peers))

				logEntry := rf.getLastLog()

				for i := range rf.NextIndex {
					rf.NextIndex[i] = logEntry.Index + 1
					rf.MatchIndex[i] = 0
				}

				go func() {
					for rf.sendAppendEntriesAll() {

						time.Sleep(time.Duration(heart) * time.Millisecond)
					}
				}()
			}
		}(i)

	}
}

//func (rf *Raft)dowhile() {
//
//	args := make(chan RequestVoteArgs)
//}

func getVoteTimeout() {
	time.Sleep(time.Duration(1500+rand.Int31n(1500)) * time.Millisecond)
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[Me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent State, and also initially holds the most
// recent saved State, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.mu = sync.Mutex{}
	rf.cond = sync.Cond{L: &rf.mu}
	rf.peers = peers
	rf.persister = persister
	rf.Me = me
	rf.CurrentTerm = 0
	rf.VotedFor = -1
	rf.CommitIndex = 0
	rf.LastApplied = 0
	rf.applyCh = applyCh
	//rf.Log = append(rf.Log, LogEntry{})
	rf.Log = make([]LogEntry, 0)
	rf.State = Followers

	rf.heartChan = make(chan AppendEntriesArgs, len(rf.peers))
	rf.snapChan = make(chan InstallSnapshotArgs, len(rf.peers))
	// Your initialization code here (2A, 2B, 2C).

	// initialize from State persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	rf.snapshot = persister.ReadSnapshot()
	DPrintf("start [%v] at [%v]", rf.Me, rf.CurrentTerm)
	// start ticker goroutine to start elections
	go rf.ticker()
	go rf.commitWorker()
	for i := 0; i < len(rf.peers); i++ {

		go func(x int) {
			for !rf.killed() {
				var heart AppendEntriesArgs
				var snap = InstallSnapshotArgs{}
				//now := time.Now()

				select {
				case snap = <-rf.snapChan:
					//befor := time.Now()
					go rf.sendSnap(snap)

				case heart = <-rf.heartChan:


					go func() {
						//befor := time.Now()
						rf.sendSingle(heart)
						//DPrintf("me[%v] x[%v] sendSingle cost [%v]", me, x, time.Now().Sub(befor))
						//DPrintf("x[%v] sendSingle cost [%v]", x, time.Now().Sub(befor))
					}()
				}

				//DPrintf("x[%v] cost [%v]", x, time.Now().Sub(now))
			}

		}(i)
	}

	return rf
}
